package server;

import java.io.BufferedInputStream;
import java.io.BufferedOutputStream;
import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.DataInputStream;
import java.io.DataOutputStream;
import java.io.IOException;
import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.net.ServerSocket;
import java.net.Socket;
import java.nio.channels.ServerSocketChannel;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import command.ExceptionResponseFactory.ExceptionResponse;
import command.ICommand;
import command.ICommandFactory;
import command.protobuf.MessageBufProtoBuf.RpcMessage;
import config.Configure;

import util.NetworkUtil;

public class RpcServer {
	public static final Logger logger = LoggerFactory
			.getLogger(RpcServer.class);

	ServerSocket rpcSocket;
	ExecutorService executor = Executors.newCachedThreadPool();
	private static RpcServer server = new RpcServer();

	public static RpcServer instance() {
		return server;
	}

	public void initServer() {
		ServerSocketChannel serverChannel;
		try {
			serverChannel = ServerSocketChannel.open();
			rpcSocket = serverChannel.socket();
			rpcSocket.setReuseAddress(true);
			InetAddress listenAddr = InetAddress.getByName(Configure
					.getRpcAddress());
			InetSocketAddress address = new InetSocketAddress(listenAddr,
					Configure.getRpcPort());

			rpcSocket.bind(address);
			new RpcSocketServerThread(rpcSocket, "inernal rpc").start();
			logger.info("rpcserver started on " + listenAddr.toString());
		} catch (IOException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}

	}

	public class RpcSocketServerThread extends Thread {
		ServerSocket serverSocket = null;

		public RpcSocketServerThread(ServerSocket sock, String name) {
			super(name);
			this.serverSocket = sock;
		}

		@Override
		public void run() {
			while (true) {
				try {
					Socket sock = serverSocket.accept();
					executor.execute(new RpcSocketWorker(sock));
				} catch (IOException e) {
					// TODO Auto-generated catch block
					e.printStackTrace();
				}
			}
		}
	}

	public static class RpcSocketWorker implements Runnable {
		private Socket sock = null;
		private ByteArrayOutputStream outArr = new ByteArrayOutputStream();

		public RpcSocketWorker(Socket sock) {
			this.sock = sock;
		}

		@Override
		public void run() {
			boolean expRaised = true;
			DataOutputStream output = null;
			RpcMessage message = null;
			try {
				DataInputStream input = new DataInputStream(
						new BufferedInputStream(sock.getInputStream(), 4096));
				output = new DataOutputStream(new BufferedOutputStream(
						sock.getOutputStream(), 4096));
				sock.setKeepAlive(true);
				sock.setTcpNoDelay(true);

				while (true) {
					expRaised = true;
					// read the command
					 message = RpcMessage
							.parseFrom(new ByteArrayInputStream(NetworkUtil
									.read(input)));

					// execute the command
					ICommandFactory factory = CommandFactories
							.getCommand(CommandFactories.cmdTypes[message
									.getHeader().getCmdType()]);
					ICommand cmd = factory.createCommand();
					cmd.setOrigin(message, sock);
					RpcMessage ret = cmd.invoke();
					// send the result
					ret.writeTo(outArr);
					NetworkUtil.write(output, outArr.toByteArray());
					expRaised = false;
					outArr.reset();
				}
			} catch (IOException e) {
				// exception raised
				if (expRaised && sock.isConnected()) {
					try {
						NetworkUtil.write(output, NetworkUtil.makeExResMessage(message, e.getMessage()));
					} catch (IOException e1) {
						//we can only close the sock now
					}
					// NetworkUtil.write(output, content);
				} else
					// it means this connection has been closed by the clients
					logger.info(sock.toString() + " closed");
			}
		}
	}
}
